package com.gitee.sparta.core.cluster;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.GlobalThreadPool;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import com.gitee.sparta.common.IdentityEnum;
import com.gitee.sparta.common.helpers.IsolationHelper;
import com.gitee.sparta.common.helpers.JsonHelper;
import com.gitee.sparta.core.bean.cluster.ClusterCmd;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.bean.cluster.ServerBean;
import com.gitee.sparta.core.cluster.balance.LoadBalance;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

/**
 * <p>
 * leader 抽象
 * </p>
 *
 * @author zyred
 * @since v 0.0.2
 */
@Slf4j
public abstract class AbstractClusterLeader {

    public static final BlockingQueue<ClusterProtocol> SEND_FOLLOWER_QUEUE = new LinkedBlockingQueue<>();

    protected final SpartaCoreProperties properties;
    protected final RedissonClient client;
    protected final LoadBalance loadBalance;

    public AbstractClusterLeader(SpartaCoreProperties properties,
                                 RedissonClient client,
                                 LoadBalance loadBalance) {
        this.properties = properties;
        this.client = client;
        this.loadBalance = loadBalance;
        this.processorCmd();
    }

    /**
     * 创建服务
     */
    abstract public void buildLeader();

    /**
     * TODO 真实连接之前，建议启动一个临时客户端去尝试一次连接，成功后再关闭自己
     * 从 leader 降级为 follower
     * <p>
     * 场景：
     * A B C D 四个服务为例，A 服务 B 服务启动成功，A 为 leader 节点，B 为 follower 节点
     * 此时构建器一个完整的集群，此时 redis 中 server_table 中的数据被删掉了，正好 C 启动
     * C 启动阶段到 redis server_table 中找不到 A 和 B 的信息，就会出现选择自己成为 leader
     * 于是，整个集群中，A 和 C 两个 leader，在这种情况下，C 肯定是不知道任何服务存在的，就
     * 需要 A 服务一直刷新 server_table 中的内容，当然 C 也一直在刷自己的信息到 server_table
     * 中，突然 A 或者 C 发现了 server_table 中存在两个 leader 节点，那么谁发现这个问题，谁
     * 就通过本方法执行服务降级。
     * <p>
     * 原理：
     * 1. 关闭搬运线程
     * 2. 关闭 leader 线程
     * 3. 广播给 此leader 下的 follower，让他们换 leader
     * 4. 关闭 server
     * 5. 连接新的 leader
     *
     * @param ip   候选者的ip地址
     * @param port 候选者的 port
     * @throws Exception 变更 follower
     */
    abstract public void demotionFollower(String ip, int port) throws Exception;

    /**
     * 实际写出数据到 follower 节点
     *
     * @param protocol 协议对象
     * @throws Exception 异常
     */
    abstract public void doWriteFollower(ClusterProtocol protocol) throws Exception;

    /**
     * 向队列中写数据
     *
     * @param protocol 协议
     */
    public static void sendQueue(ClusterProtocol protocol) {
        SEND_FOLLOWER_QUEUE.add(protocol);
    }


    /**
     * 启动一个线程,读取阻塞队列中的数据, 根据指令做处理
     */
    public void processorCmd() {
        GlobalThreadPool.execute(() -> {
            while (!Thread.interrupted()) {
                ClusterProtocol protocol = null;
                try {
                    protocol = SEND_FOLLOWER_QUEUE.take();
                    final byte cmd = protocol.getCmd();

                    // 内部处理的指令
                    if (ClusterCmd.isLeaderInnerCmd(cmd)) {
                        if (cmd == ClusterCmd.FLUSH_SERVER_TABLE) {
                            // 刷新 table
                            this.flushServerTable();
                            continue;
                        }
                        // 移除负载均衡
                        this.removeLoadBalance(protocol);
                        // 自身为 follower，需要关闭本上的 leader 线程
                        this.stopLeaderSendThread(cmd);
                        continue;
                    }

                    // 往 follower 发送的指令
                    this.doWriteFollower(protocol);
                } catch (Exception e) {
                    log.error("[Sparta] Send follower, protocol: {}, error: {}",
                            JsonHelper.getJsonString(protocol), e.getMessage(), e);
                    if (Objects.nonNull(protocol)) {
                        SEND_FOLLOWER_QUEUE.add(protocol);
                    }
                }
            }
        });
    }

    /**
     * 服务启动阶段是无法区分自身是 follower 还是 leader
     * 但是，启动阶段需要启动发送消息的线程，所以这里如果
     * 是 follower，就需要停止 leader 发送消息的线程
     *
     * @param cmd 指令
     */
    private void stopLeaderSendThread(byte cmd) {
        if (cmd == ClusterCmd.STOP_LEADER_SENDER) {
            Thread.currentThread().interrupt();
            log.info("[Sparta] Stop leader thread ...");
        }
    }

    /**
     * follower 超时，移除 follower 的负载权限
     *
     * @param protocol 协议
     */
    private void removeLoadBalance(ClusterProtocol protocol) {
        if (protocol.getCmd() != ClusterCmd.REMOVE_LOAD_BALANCE) {
            return;
        }

        String isolation = IsolationHelper.isolation(protocol.getServerId());
        if (this.loadBalance.contains(isolation)) {
            this.loadBalance.removeNode(protocol.getServerId());
            log.info("[Sparta] Load balance removed, sid: {}", protocol.getServerId());
        }
    }


    /**
     * server_table 刷新
     * <pre>
     * </pre>
     */
    private void flushServerTable() {
        Map<Integer, ServerBean> localTable = new HashMap<>(ClusterCache.getServerTable());
        RMap<Integer, ServerBean> serverTable = this.client.getMap(this.properties.getServerTable());

        ServerBean self = localTable.get(ClusterCache.getServerId());

        if (!localTable.equals(serverTable)) {
            RLock lock = this.client.getLock(this.properties.getDemotionLock());
            try {
                lock.lock();
                int counter = 0, maxCounter = 5;
                // serverTable 被人为删除，需要将 leader 中的数据刷写到 serverTable 中
                // 以 localTable 中的数据为准
                if (CollUtil.isEmpty(serverTable)) {
                    self.setCandidate(true);
                    localTable.remove(self.getServerId());
                    localTable.put(self.getServerId(), self);
                    serverTable.putAll(localTable);
                    return;
                }

                // 从 redis 服务列表中得到一个后选择，但候选者能够正常被连接
                if (this.filtrateHitCandidate(serverTable, self, maxCounter)) {
                    return;
                }

                Set<ServerBean> leaders = serverTable.values().stream()
                        .filter(ServerBean::isLeader).collect(Collectors.toSet());

                // 如果有多个 leader， 则选择出一个的 leader
                ServerBean candidate = this.selectCandidateLeader(leaders, self);
                if (!candidate.equals(self)) {
                    while (StrUtil.isNotBlank(candidate.getServerIp())
                            || candidate.getServerPort() != 0) {

                        if (counter >= maxCounter) {
                            candidate = self;
                            break;
                        }
                        candidate = this.selectCandidateLeader(leaders, self);
                        counter++;
                    }
                }

                candidate.setCandidate(true);
                // 如果自己被选中，那么就让别人来连接自己, 还是以自己的数据为准
                if (candidate.equals(self)) {
                    serverTable.clear();
                    localTable.put(self.getServerId(), self);
                    serverTable.putAll(localTable);
                    serverTable.putAll(localTable);
                    log.info("[Sparta] Candidate self : {}", JsonHelper.getJsonString(self));
                    return;
                }

                this.selfFailTobeElected(serverTable, self, candidate);
            } finally {
                lock.unlock();
            }
        }
    }


    /**
     * 筛选命中的候选者
     *
     * @param st         redis 中的服务列表
     * @param self       leader 自己
     * @param maxCounter 最大次数
     * @return true: 候选成功，false: 候选失败
     */
    private boolean filtrateHitCandidate(RMap<Integer, ServerBean> st, ServerBean self, int maxCounter) {
        ServerBean sc = st.values().stream().filter(ServerBean::isCandidate).findFirst().orElse(null);

        if (Objects.nonNull(sc) && !sc.equals(self)) {

            int counter = 0;
            while (StrUtil.isBlank(sc.getServerIp())
                    || sc.getServerPort() == 0) {
                if (counter >= maxCounter) {
                    sc = self;
                    break;
                }

                sc.setCandidate(false);
                st.remove(sc.getServerId());
                st.put(sc.getServerId(), sc);
                sc = st.values().stream().filter(ServerBean::isCandidate)
                        .findFirst().orElse(null);

                if (Objects.isNull(sc)) {
                    sc = self;
                    break;
                }
                counter++;
            }

            boolean isSelf = !sc.equals(self);
            if (isSelf) {
                ClusterCache.setServerIdentity(IdentityEnum.FOLLOWER);
                ClusterCache.setLeaderSid(sc.getServerId());
                ClusterCache.demotionFollower();

                log.info("[Sparta] Find candidate, connect candidate : {}",
                        JsonHelper.getJsonString(sc));

                ServerBean target = sc;
                GlobalThreadPool.execute(() -> {
                    try {
                        this.demotionFollower(target.getServerIp(), target.getServerPort());
                    } catch (Exception e) {
                        log.error("[Sparta] Demotion follower fail: {}", e.getMessage(), e);
                    }
                });
            }
            return isSelf;
        }
        return false;
    }

    /**
     * self 未能当选 leader
     *
     * @param st        redis 服务列表
     * @param self      leader 自己
     * @param candidate 候选者
     */
    private void selfFailTobeElected(RMap<Integer, ServerBean> st, ServerBean self, ServerBean candidate) {
        log.info("[Sparta] Candidate other : {}", JsonHelper.getJsonString(candidate));
        self.setLeader(false);
        self.setCandidate(false);
        self.setServerPort(0);

        st.remove(self.getServerId());
        st.put(self.getServerId(), self);
        st.remove(candidate.getServerId());
        st.put(candidate.getServerId(), candidate);

        ClusterCache.setServerIdentity(IdentityEnum.FOLLOWER);
        ClusterCache.setLeaderSid(candidate.getServerId());
        ClusterCache.demotionFollower();

        GlobalThreadPool.execute(() -> {
            try {
                this.demotionFollower(candidate.getServerIp(), candidate.getServerPort());
            } catch (Exception e) {
                log.error("[Sparta] Demotion follower fail: {}", e.getMessage(), e);
            }
        });
    }

    /**
     * 从一堆 leader 中选择出一个候选者 leader
     *
     * @param leaders 一群 leader
     * @param self    发现 local 与 server table 中不一样的那个 leader
     * @return 选择出一个非自己的候选者
     */
    private ServerBean selectCandidateLeader(Set<ServerBean> leaders, ServerBean self) {
        if (leaders.size() == 0) {
            return self;
        }

        Set<ServerBean> temp = new HashSet<>(leaders);
        temp.remove(self);

        ArrayList<ServerBean> arr = new ArrayList<>(temp);
        if (arr.size() == 0) {
            return self;
        }
        if (arr.size() == 1) {
            return arr.get(0);
        }
        return arr.get(RandomUtil.randomInt(arr.size()));
    }


}
